package com.atguigu.tran;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;
import java.util.List;

/**
 * @author yhm
 * @create 2024-04-02 9:20
 */
public class Test01_Map {
    public static void main(String[] args) throws Exception {
        // 1. 创建环境
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 8081);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        // 2. 读取数据源
        List<WaterSensor> waterSensors = Arrays.asList(new WaterSensor("sensor_1", 1000L, 10),
                new WaterSensor("sensor_2", 2000L, 20),
                new WaterSensor("sensor_3", 3000L, 30),
                new WaterSensor("sensor_4", 4000L, 40)
        );
        DataStreamSource<WaterSensor> dataStreamSource = env.fromCollection(waterSensors);

        // 3. 处理数据
//        SingleOutputStreamOperator<String> outputStreamOperator = dataStreamSource.map(
//                //匿名子类
//        new MapFunction<WaterSensor, String>() {
//            @Override
//            public String map(WaterSensor value) throws Exception {
//                return value.id;
//            }
//        });
        // lambda表达式
        // 容易出现类型丢失的问题
        // 如果出现类型丢失报错  需要收到添加类型
        SingleOutputStreamOperator<String> outputStreamOperator = dataStreamSource.map((WaterSensor waterSensor) -> waterSensor.id).returns(Types.STRING);

        SingleOutputStreamOperator<String> outputStreamOperator1 = dataStreamSource.map( waterSensor -> waterSensor.id);

        // 使用子类来实现对应的接口
        dataStreamSource.map(new CustomMapFunction());
        // 4. 输出
        outputStreamOperator.print();

        // 5. 执行环境
        env.execute();
    }

    // 使用子类实现对应的接口或继承对应的父类
    // 用于相对比较复杂的情况
    public static class CustomMapFunction implements MapFunction<WaterSensor,String>{

        @Override
        public String map(WaterSensor value) throws Exception {
            return value.id;
        }
    }
}
